/*
 * Copyright 2021-2022 Open Kunlun Technology <https://www.openkunlun.io>
 */

package io.openkunlun.scaladsl.serialization

import akka.actor.{ ExtendedActorSystem, Extension, ExtensionId, ExtensionIdProvider }
import akka.serialization.SerializationExtension

/**
 * @author ericxin.
 */
object AkkaSerialization extends ExtensionId[AkkaSerialization] with ExtensionIdProvider {
  override def lookup: ExtensionId[AkkaSerialization] = AkkaSerialization
  override def createExtension(system: ExtendedActorSystem) = new AkkaSerialization(system)
}

class AkkaSerialization(system: ExtendedActorSystem) extends Extension with DaprObjectSerialization {
  val serialization = SerializationExtension(system)

  /**
   *
   * @param serializable
   * @return
   */
  override def serialize[T](serializable: T): Array[Byte] = {
    val o = serializable.asInstanceOf[AnyRef]
    val serializer = serialization.findSerializerFor(o)
    serializer.toBinary(o)
  }

  /**
   *
   * @param bytes
   * @tparam T
   * @return
   */
  override def deserialize[T: Manifest](bytes: Array[Byte]): T = {
    val clazz = system.dynamicAccess.getClassFor[T](manifest[T].toString()).get
    serialization.deserialize(bytes, clazz).get
  }
}
